package com.zw.cdc;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class MySqlBinlogSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder().hostname("192.168.1.226")
            .port(3306)
            .databaseList("lzw") // set captured database
            .tableList("lzw.t1") // set captured table
            .username("root")
            .password("123456")
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
    
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    env.setParallelism(1);
//    env.enableCheckpointing(5000);
//    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//    env.getCheckpointConfig().setCheckpointTimeout(1800000L);
//    env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//    env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000L);
//    env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);
    
    env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source1").addSink(new PrintSinkFunction<String>());
    
    env.execute("Print MySQL Snapshot + Binlog1");
  }
}
